package com.thiswhyme.republish.rabbitmq.datasource;

import com.thiswhyme.common.expection.BusinessException;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @description: 历史MessageInfo仓储服务
 * @author: thiswhyme
 * @date: 2021/3/29
 */
public class HistoryMessageInfoRepo {

    public static final Logger log = LoggerFactory.getLogger(HistoryMessageInfoRepo.class);

    private DataSource dataSource;

    private MessageInfoSqlBuilder messageInfoSqlBuilder;

    public HistoryMessageInfoRepo(DataSource ds, MessageInfoSqlBuilder misb) {
        this.dataSource = ds;
        this.messageInfoSqlBuilder = misb;
    }

    /**
     * 批量迁移消息数据到历史表
     * @param msgs 待迁移的数据
     * @param suffix 分表后缀
     */
    public void batchTransfer(List<MessageInfo> msgs, String suffix) {
        Connection conn = null;
        PreparedStatement pst = null;
        List<Long> dataIds = msgs.stream().map(MessageInfo::getId).collect(Collectors.toList());
        try {
            conn = dataSource.getConnection();
            conn.setAutoCommit(false);

            //历史表新增
            String sql = messageInfoSqlBuilder.buildInsertSql(suffix);
            pst = conn.prepareStatement(sql);
            for (MessageInfo msg : msgs) {
                pst.setLong(1, msg.getId());
                pst.setInt(2, msg.getType());
                pst.setString(3, msg.getChannelName());
                pst.setString(4, msg.getMsgId());
                pst.setString(5, msg.getHeaders());
                pst.setString(6, msg.getContent());
                pst.setString(7, msg.getContentClassType());
                pst.setInt(8, msg.getPublishStatus());
                pst.setInt(9, msg.getRetryTimes());
                pst.setTimestamp(10, msg.getCreateTime());
                pst.setTimestamp(11, msg.getUpdateTime());
                pst.addBatch();
            }
            pst.executeBatch();

            //实时表删除
            sql = messageInfoSqlBuilder.buildDeleteSql(dataIds, "");
            pst = conn.prepareStatement(sql);
            for (int i = 1; i <= dataIds.size(); i++) {
                pst.setLong(i, dataIds.get(i - 1));
            }
            pst.executeUpdate();

            //提交事务
            conn.commit();
        } catch (Exception ex) {
            try {
                conn.rollback();
            } catch (SQLException e) {
                log.error("batchTransfer回滚失败", ex);
                throw new BusinessException("消息迁移失败");
            }
            log.error("执行batchTransfer失败", ex);
            throw new BusinessException("消息迁移失败");
        } finally {
            try {
                conn.setAutoCommit(true);
                if (pst != null) {
                    pst.close();
                }
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException closeEx) {
                log.warn("关闭JDBC连接失败 ... ", closeEx);
            }
        }
    }

    /**
     * 批量删除
     * @param dataIds
     * @return
     */
    public int batchDelete(List<Long> dataIds, String suffix) {
        Connection conn = null;
        PreparedStatement pst = null;
        try {
            conn = dataSource.getConnection();
            String sql = messageInfoSqlBuilder.buildDeleteSql(dataIds, suffix);
            pst = conn.prepareStatement(sql);
            for (int i = 1; i <= dataIds.size(); i++) {
                pst.setLong(i, dataIds.get(i - 1));
            }
            return pst.executeUpdate();
        } catch (Exception ex) {
            log.error("执行历史delete失败，参数：{}", dataIds, ex);
            throw new BusinessException("批量删除消息失败");
        } finally {
            try {
                if (pst != null) {
                    pst.close();
                }
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException closeEx) {
                log.warn("关闭JDBC连接失败 ... ", closeEx);
            }
        }
    }

    /**
     * 获取历史分表名
     * @param time
     * @return
     */
    public String generateTableNameSuffix(Timestamp time) {
        String month = DateFormatUtils.format(time, "MM");
        return "_" + month;
    }
}
